#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/log.hpp"
#include "../mqcommon/mq_meg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadpool.hpp"

#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include "mq_route.hpp"

namespace pprmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using OpenChannelRequestPtr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<CloseChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string& cid,
                const muduo::net::TcpConnectionPtr& conn,
                const ProtobufCodecPtr& codec,
                const ConsumerManager::ptr& cmp,
                const threadpool::ptr& pool,
                const VirtualHost::ptr& host)
                :_cid(cid),_conn(conn),_codec(codec)
                ,_cmp(cmp),_pool(pool),_host(host)
        {}
        ~Channel() 
        {
            if (_consumer.get() != nullptr) 
            {
                _cmp->remove(_consumer->_tag, _consumer->_qname);
            }
        }
        // 交换机的声明和删除
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(),req->exchange_type(),
                                req->durable(),req->auto_delete(),req->args());
            return basicResponse(ret,req->rid(),req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        // 队列的声明和删除
        void declareQueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareQueue(req->queue_name(),req->durable(),req->exclusive()
                                        ,req->auto_delete(),req->args());
            if(ret == false) return basicResponse(false,req->rid(),req->cid());
            _cmp->initQueueConsumer(req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr& req)
        {
            // 清理队列的消费者管理
            _cmp->destroyQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        // 队列的绑定与删除
        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret = _host->bind(req->exchange_name(),req->queue_name(),req->binding_key());
            return basicResponse(ret,req->rid(),req->cid());
        }
        void queueUnBind(const queueUnBindRequestPtr& req)
        {
            _host->unBind(req->exchange_name(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        // 消息的发布
        void basicPublish(const basicPublishRequestPtr& req)
        {
            //1. 判断交换机是否存在
            auto ep = _host->selectExchange(req->exchange_name());
            if(ep.get() == nullptr)
            {
                return basicResponse(false,req->rid(),req->cid());
            }
            //2. 交换路由，确定目标队列，判断消息可以发布到交换机绑定的哪个队列中
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties* properties = nullptr;
            std::string routing_key;
            if(req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = req->mutable_properties()->routing_key();
            }
            for(auto& binding : mqbm)
            {
                if(Route::route(ep->_type,routing_key,binding.second->_binding_key))
                {
                    //3. 将消息添加到队列中
                    _host->basicPublish(binding.first,properties,req->body());
                    //4. 向线程池中添加一个消息消费任务（向指定队列的订阅者去推送消息--线程池完成）
                    auto task = std::bind(&Channel::consume,this,binding.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true,req->rid(),req->cid());
        }
        // 消息的确认
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(),req->message_id());
            return basicResponse(true,req->rid(),req->cid());
        }
        // 订阅队列消息
        void basicConsume(const basicConsumeRequestPtr& resp)
        {
            //1. 判断队列是否存在
            bool ret = _host->existsQueue(resp->queue_name());
            if(ret == false)
            {
                DLOG("队列不存在 %s", resp->queue_name().c_str());
                return basicResponse(false,resp->rid(),resp->cid());
            }
            //2. 创建消费者
            auto bp = std::bind(&Channel::callback,this,std::placeholders::_1,std::placeholders::_2, std::placeholders::_3);
            _consumer = _cmp->create(resp->consumer_tag(),resp->queue_name(),resp->auto_ack(),bp);

            return basicResponse(true,resp->rid(),resp->cid());
        }
        // 取消订阅
        void basicCancel(const basicCancelRequestPtr& req)
        {
            _cmp->remove(req->consumer_tag(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }
    private:
        // 消息推送回调
        // 当队列有消息时,通过该回调推送给客户端
        void callback(const std::string tag, const BasicProperties *bp, const std::string &body)
        {
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if(bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery(bp->delivery());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn,resp);
        }
        // 异步消费任务
        void consume(const std::string &qname)
        {
            //1. 从队列取消息
            MessagePtr msg = _host->basicConsume(qname);
            if(msg.get() == nullptr)
            {
                DLOG("消费任务失败，%s 队列无消息", qname.c_str());
                return;
            }
            //2. 选择一个消费者
            Consumer::ptr consumer = _cmp->choose(qname);
            if(consumer.get() == nullptr)
            {
                DLOG("消费任务失败，%s 队列无消费者", qname.c_str());
                return;
            }
            //3. 调用消费者回调推送消息
            consumer->_callback(consumer->_tag,msg->mutable_payload()->mutable_properties(),msg->payload().body());
            //4. 自动确认则直接标记消息已消费
            if(consumer->_auto_ack) _host->basicAck(qname,msg->payload().properties().id());
        }
        // 构造通用的操作结果响应，统一格式便于客户端解析
        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid); // 请求ID，用于匹配请求与响应
            resp.set_cid(cid); // 信道ID，标识响应所属的信道
            resp.set_ok(ok); // 操作结果（成功/失败）
            _codec->send(_conn,resp);
        }

    private:
        std::string _cid; // 信号的唯一标识
        muduo::net::TcpConnectionPtr _conn; // muduo库的Tcp管理，网络通信
        ProtobufCodecPtr _codec; // 用于处理序列化和反序列化
        Consumer::ptr _consumer; // 当前信道关联的消费者，此时信道处理订阅状态则有效
        ConsumerManager::ptr _cmp; // 消费者管理器，管理订阅队列的消费者
        threadpool::ptr _pool; // 线程池，异步处理消息消费任务，避免网络IO线程
        VirtualHost::ptr _host; // 管理交换机，队列等核心资源
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        bool openChannel(const std::string& id,
                const muduo::net::TcpConnectionPtr& conn,
                const ProtobufCodecPtr& codec,
                const ConsumerManager::ptr& cmp,
                const threadpool::ptr& pool,
                const VirtualHost::ptr& host)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 在哈希表中查找该id的信道，判断是否已存在
            auto it = _channels.find(id);
            if(it != _channels.end())
            {
                // 存在
                return false;
            }

            // 创建新的Channel实例
            auto channel = std::make_shared<Channel>(id,conn,codec,cmp,pool,host);
            // 将新创建的Channel插入哈希表
            _channels.insert(std::make_pair(id,channel));
            return true;
        }

        // 关闭并销毁指定id的信道
        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 从哈希管理表中删除
            _channels.erase(id);
        }

        // 根据id查询信道
        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if(it == _channels.end())
            {
                // 没有找到
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::ptr> _channels;
    };
}

#endif